package com.mazha.rabbitmq.listener;

import com.mazha.rabbitmq.base.MqMessage;
import com.mazha.rabbitmq.common.cache.ChatGlobalCache;
import com.mazha.rabbitmq.pojo.dto.ChatGroupDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 聊天室消息监听
 *
 * @author leizi
 * @date 2023/12/10
 */
@Component
@RabbitListener(queues = "chat.speak.queue")
@Slf4j
public class ChatSpeakReceiver {

    private final ThreadPoolExecutor chatSpeakThreadPool;

    public ChatSpeakReceiver(@Qualifier(value = "chatSpeakThreadPool") ThreadPoolExecutor chatSpeakThreadPool) {
        this.chatSpeakThreadPool = chatSpeakThreadPool;
    }

    /**
     * 监听消息
     *
     * @param message 消息体
     */
    @RabbitHandler
    public void process(MqMessage message) {
        log.info("chat speak receiver:{}", message);
        ConcurrentHashMap<String, ChatGroupDTO> chatGroupCache = ChatGlobalCache.chatGroupCache;
        ConcurrentHashMap<Long, SseEmitter> userSseCache = ChatGlobalCache.userSseCache;

        chatSpeakThreadPool.execute(()->{
            ChatGroupDTO chatGroupDTO = chatGroupCache.get(message.getMethod());
            List<Long> userIdList = chatGroupDTO.getUserIdList();
            // 屏蔽发起人用户
            userIdList.remove(message.getShieldMethod());
            // 广播消息
            for (Long userId : userIdList) {
                SseEmitter sseEmitter = userSseCache.get(userId);
                // 同步用户消息sse
                try {
                    sseEmitter.send(message.getData());
                    // 会关闭连接
//                    sseEmitter.complete();
                } catch (IOException e) {
                    log.error("chat sse send error:{}", e);
                } catch (Exception e) {
                    sseEmitter.completeWithError(e);
                }
            }
        });

    }

}
